#include <unordered_set>
#include "OpGroupBy.h"
#include "common.h"

#include "Hasher.h"
#include "utils/logger.h"

using namespace internal;
using namespace std;


OpGroupBy::OpGroupBy(const GroupByOperation& operation) {
	
	row_size = 0;
	next_row = end_row = 0;
	is_computed_hash_set = false;
	
	source = new OpInput(operation.source());
	in = source->getRowsBuffer();
	types = source->getTypes();
	
	// przetwarzanie kolumn po ktorych grupuje
	int n = types.size();
	for(int i=0; i<n; i++) {
		switch(types[i]) {
			case INT:
				output.push_back( Output_elem(new char[sizeof(int32_t)*TABLESIZE],INT) );
				row_size += sizeof(int32_t);
				break;
			case BOOL:
				output.push_back( Output_elem(new char[sizeof(bool)*TABLESIZE],BOOL) );
				row_size += sizeof(int32_t);
				break;
			case DOUBLE:
				output.push_back( Output_elem(new char[sizeof(double)*TABLESIZE],DOUBLE) );
				row_size += sizeof(double);
				break;
			default: assert(false);
		}	
	}
	
	row_size += sizeof(size_t);			// dokladamy hash'a
	
	by_column_size = operation.group_by_column_size();
}


void OpGroupBy::allocate_next_rows() {
	next_row = new char[row_size * TABLESIZE];
	rows.push_back(next_row);
	end_row = next_row + row_size * TABLESIZE;
}

int OpGroupBy::doit() {
	if(!is_computed_hash_set) {
		is_computed_hash_set = true;
		
		typedef unordered_set<char*, IdHasher, RowEqual > RowHashSet;
		RowHashSet S(TABLESIZE, IdHasher() ,
								RowEqual (by_column_size ));
		
    RowHasher hashFunc(by_column_size);
    
		int size;
		while( (size = source->doit()) != -1) {
			char* end = in + size*(row_size - sizeof(size_t));
			for(char* next = in; next != end; next += row_size - sizeof(size_t)) {	
				if(next_row == end_row) allocate_next_rows();

				memcpy(next_row + sizeof(size_t) , next, row_size -sizeof(size_t));
			
				*((size_t*) next_row) = hashFunc(next_row);
				
				pair<RowHashSet::iterator, bool> p = S.insert(next_row);

				if(p.second) {
					next_row += row_size;
				}
				else {
					char* q = (*p.first) + sizeof(size_t) + by_column_size*sizeof(int32_t);
					char* s = next_row   + sizeof(size_t) + by_column_size*sizeof(int32_t);
				
					// wlasciwa agragacja - teraz mamy tylko sumowanie !
					for(int j=by_column_size; j<types.size(); j++) {
						switch(types[j]) {
							case INT: {
								int32_t *wsk = (int32_t*) q;
								*wsk += *((int32_t*) s);
								q += sizeof(int32_t);
								s += sizeof(int32_t);
								break;
							}
							case DOUBLE: {
								double *wsk = (double*) q;
								*wsk += *((double*) s);
								q += sizeof(double);
								s += sizeof(double);
								break;
							}
							default: assert(false);
						}
					}
				}
			}
		}
	}
	
	if (rows.empty()) {
    return -1;
  } else {
		char* q = rows.back();
		int size=0;
		while(q != next_row) {
			char* r = q;
			r += sizeof(size_t);		// pomijamy hash'a
			for(int j=0; j<output.size(); j++) {
				switch(output[j].type) {
					case INT:
						*(((int32_t*) output[j].column) + size) = *( (int32_t*)r );
						r += sizeof(int32_t);
						break;
					case DOUBLE:
						*(((double*) output[j].column) + size) = *( (double*)r );
						r += sizeof(double);
						break;
					case BOOL:
						*(((bool*) output[j].column) + size) = (bool) *( (int32_t*)r );
						r += sizeof(int32_t);
						break;
				}
			}
			size++;
			q += row_size;
		}
		delete [] (rows.back());
		rows.pop_back();
		if (!rows.empty()) {
      next_row = rows.back() + row_size*TABLESIZE;
    }
		return size;
	}
}

